package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.util.MykafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @ClassName dwdTradeOrderAdd
 * @Description TODO 下单人数
 * @Author AFei
 * @Date 2022/9/6 0:56
 * @Version 1.0
 */
public class dwdTradeOrderAdd {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql(MykafkaUtil.getTopicDDL("dwd_trade_order_add"));

        // TODO 从业务数据中筛选出下单数据
        Table orderAdd = tableEnv.sqlQuery("select\n" +
                "        data['id'] id,\n" +
                "        data['user_id'] user_id,\n" +
                "        data['origin_amount'] origin_amount,\n" +
                "        data['coupon_reduce'] coupon_reduce,\n" +
                "        data['final_amount'] final_amount,\n" +
                "        ts,\n" +
                "        proc_time\n" +
                "        from topic_db where `table`='order_info' and \n" +
                "        (`type` = 'insert' or `type` = 'update')");
        tableEnv.createTemporaryView("order_add",orderAdd);
        //tableEnv.executeSql("select * from order_add").print();
        // TODO 将关联的结果写到kafka主题中
        // 创建动态表和要写入的主题进行映射
        tableEnv.executeSql("CREATE TABLE dwd_trade_order_add ( \n" +
                "                  id string, \n" +
                "                  user_id string, \n" +
                "                  origin_amount string, \n" +
                "                  coupon_reduce string, \n" +
                "                  final_amount string, \n" +
                "                  ts string, \n" +
                "                  proc_time TIMESTAMP_LTZ(3), \n" +
                "                  PRIMARY KEY (id) NOT ENFORCED \n" +
                "                )" + MykafkaUtil.getUpsert("dwd_trade_order_add"));

        tableEnv.executeSql("insert into dwd_trade_order_add select * from order_add");
    }
}
